package cn.dataling.dag.test;

import cn.dataling.dag.enums.DagWorkflowNodeState;
import cn.dataling.dag.execute.DagWorkflowExecutor;
import cn.dataling.dag.pojo.DagWorkflow;
import cn.dataling.dag.pojo.DagWorkflowEdge;
import cn.dataling.dag.pojo.DagWorkflowNode;
import cn.dataling.dag.task.FlinkDagWorkflowTask;
import cn.dataling.dag.util.ThreadPoolUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;

import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.stream.Collectors;

public class Main2 {
    public static void main(String[] args) {
        String jobGraphData = "" +
                "{\"cells\":[{\"position\":{\"x\":-540,\"y\":-230},\"size\":{\"width\":180,\"height\":70},\"view\":\"vue-shape-view\",\"shape\":\"job-node\",\"id\":\"82d242d8-da39-48f0-9399-986c6604c289\",\"component\":\"job-node\",\"data\":{\"name\":\"数据开发\",\"jobStatus\":\"未提交\",\"nodeType\":\"SQL\",\"taskName\":\"数据目录\",\"taskId\":10,\"nodeId\":\"82d242d8-da39-48f0-9399-986c6604c289\"},\"zIndex\":1,\"ports\":{\"groups\":{\"in\":{\"attrs\":{\"circle\":{\"r\":5,\"magnet\":true,\"strokeWidth\":2,\"fill\":\"transparent\",\"stroke\":\"transparent\"}},\"position\":{\"name\":\"left\"}},\"out\":{\"attrs\":{\"circle\":{\"r\":5,\"magnet\":true,\"strokeWidth\":2,\"fill\":\"transparent\",\"stroke\":\"transparent\"}},\"position\":{\"name\":\"right\"}}},\"items\":[{\"id\":\"in\",\"group\":\"in\",\"attrs\":{\"circle\":{\"fill\":\"transparent\",\"stroke\":\"transparent\"}}},{\"id\":\"out\",\"group\":\"out\",\"attrs\":{\"circle\":{\"fill\":\"transparent\",\"stroke\":\"transparent\"}}}]}},{\"position\":{\"x\":-241,\"y\":-230},\"size\":{\"width\":180,\"height\":70},\"view\":\"vue-shape-view\",\"shape\":\"job-node\",\"id\":\"d91cbac7-dcec-4044-ac00-e62ac709d995\",\"component\":\"job-node\",\"data\":{\"name\":\"数据ETL\",\"jobStatus\":\"未提交\",\"nodeType\":\"GRAPH\",\"taskName\":\"数据源同步\",\"taskId\":11,\"nodeId\":\"d91cbac7-dcec-4044-ac00-e62ac709d995\"},\"zIndex\":2,\"ports\":{\"groups\":{\"in\":{\"attrs\":{\"circle\":{\"r\":5,\"magnet\":true,\"strokeWidth\":2,\"fill\":\"transparent\",\"stroke\":\"transparent\"}},\"position\":{\"name\":\"left\"}},\"out\":{\"attrs\":{\"circle\":{\"r\":5,\"magnet\":true,\"strokeWidth\":2,\"fill\":\"transparent\",\"stroke\":\"transparent\"}},\"position\":{\"name\":\"right\"}}},\"items\":[{\"id\":\"in\",\"group\":\"in\",\"attrs\":{\"circle\":{\"fill\":\"transparent\",\"stroke\":\"transparent\"}}},{\"id\":\"out\",\"group\":\"out\",\"attrs\":{\"circle\":{\"fill\":\"transparent\",\"stroke\":\"transparent\"}}}]}},{\"position\":{\"x\":70,\"y\":-230},\"size\":{\"width\":180,\"height\":70},\"view\":\"vue-shape-view\",\"shape\":\"job-node\",\"id\":\"131fe4c4-9b21-4cbd-9663-873c4795041b\",\"component\":\"job-node\",\"data\":{\"name\":\"数据ETL\",\"jobStatus\":\"未提交\",\"nodeType\":\"GRAPH\",\"taskName\":\"数据生成\",\"taskId\":12,\"nodeId\":\"131fe4c4-9b21-4cbd-9663-873c4795041b\"},\"zIndex\":3,\"ports\":{\"groups\":{\"in\":{\"attrs\":{\"circle\":{\"r\":5,\"magnet\":true,\"strokeWidth\":2,\"fill\":\"transparent\",\"stroke\":\"transparent\"}},\"position\":{\"name\":\"left\"}},\"out\":{\"attrs\":{\"circle\":{\"r\":5,\"magnet\":true,\"strokeWidth\":2,\"fill\":\"transparent\",\"stroke\":\"transparent\"}},\"position\":{\"name\":\"right\"}}},\"items\":[{\"id\":\"in\",\"group\":\"in\",\"attrs\":{\"circle\":{\"fill\":\"transparent\",\"stroke\":\"transparent\"}}},{\"id\":\"out\",\"group\":\"out\",\"attrs\":{\"circle\":{\"fill\":\"transparent\",\"stroke\":\"transparent\"}}}]}},{\"shape\":\"edge\",\"attrs\":{\"line\":{\"stroke\":\"#4e6ef2\",\"strokeDasharray\":5,\"style\":{\"animation\":\"ant-line 30s infinite linear\"}}},\"id\":\"e15b3c22-80e2-42ac-908f-ac03f8f04284\",\"connector\":{\"name\":\"smooth\"},\"zIndex\":4,\"source\":{\"cell\":\"82d242d8-da39-48f0-9399-986c6604c289\",\"port\":\"out\"},\"target\":{\"cell\":\"d91cbac7-dcec-4044-ac00-e62ac709d995\",\"port\":\"in\"}},{\"shape\":\"edge\",\"attrs\":{\"line\":{\"stroke\":\"#4e6ef2\",\"strokeDasharray\":5,\"style\":{\"animation\":\"ant-line 30s infinite linear\"}}},\"id\":\"0c2d658b-4b5a-4886-aec1-b94ba9f75d12\",\"connector\":{\"name\":\"smooth\"},\"zIndex\":5,\"source\":{\"cell\":\"d91cbac7-dcec-4044-ac00-e62ac709d995\",\"port\":\"out\"},\"target\":{\"cell\":\"131fe4c4-9b21-4cbd-9663-873c4795041b\",\"port\":\"in\"}},{\"position\":{\"x\":-540,\"y\":-69},\"size\":{\"width\":180,\"height\":70},\"view\":\"vue-shape-view\",\"shape\":\"job-node\",\"id\":\"65dcb08a-15c7-4dd3-a46f-f75326bde3e6\",\"component\":\"job-node\",\"data\":{\"name\":\"数据开发\",\"jobStatus\":\"未提交\",\"nodeType\":\"SQL\",\"taskName\":\"数据目录\",\"taskId\":10,\"nodeId\":\"65dcb08a-15c7-4dd3-a46f-f75326bde3e6\"},\"ports\":{\"groups\":{\"in\":{\"attrs\":{\"circle\":{\"r\":5,\"magnet\":true,\"strokeWidth\":2,\"fill\":\"transparent\",\"stroke\":\"transparent\"}},\"position\":{\"name\":\"left\"}},\"out\":{\"attrs\":{\"circle\":{\"r\":5,\"magnet\":true,\"strokeWidth\":2,\"fill\":\"transparent\",\"stroke\":\"transparent\"}},\"position\":{\"name\":\"right\"}}},\"items\":[{\"id\":\"in\",\"group\":\"in\",\"attrs\":{\"circle\":{\"fill\":\"transparent\",\"stroke\":\"transparent\"}}},{\"id\":\"out\",\"group\":\"out\",\"attrs\":{\"circle\":{\"fill\":\"transparent\",\"stroke\":\"transparent\"}}}]},\"zIndex\":6},{\"position\":{\"x\":-241,\"y\":-69},\"size\":{\"width\":180,\"height\":70},\"view\":\"vue-shape-view\",\"shape\":\"job-node\",\"id\":\"023d6210-78e1-4bec-979d-53a7ca5abbc6\",\"component\":\"job-node\",\"data\":{\"name\":\"数据ETL\",\"jobStatus\":\"未提交\",\"nodeType\":\"GRAPH\",\"taskName\":\"数据源同步\",\"taskId\":11,\"nodeId\":\"023d6210-78e1-4bec-979d-53a7ca5abbc6\"},\"ports\":{\"groups\":{\"in\":{\"attrs\":{\"circle\":{\"r\":5,\"magnet\":true,\"strokeWidth\":2,\"fill\":\"transparent\",\"stroke\":\"transparent\"}},\"position\":{\"name\":\"left\"}},\"out\":{\"attrs\":{\"circle\":{\"r\":5,\"magnet\":true,\"strokeWidth\":2,\"fill\":\"transparent\",\"stroke\":\"transparent\"}},\"position\":{\"name\":\"right\"}}},\"items\":[{\"id\":\"in\",\"group\":\"in\",\"attrs\":{\"circle\":{\"fill\":\"transparent\",\"stroke\":\"transparent\"}}},{\"id\":\"out\",\"group\":\"out\",\"attrs\":{\"circle\":{\"fill\":\"transparent\",\"stroke\":\"transparent\"}}}]},\"zIndex\":7},{\"shape\":\"edge\",\"attrs\":{\"line\":{\"stroke\":\"#4e6ef2\",\"strokeDasharray\":5,\"style\":{\"animation\":\"ant-line 30s infinite linear\"}}},\"id\":\"4dc09357-cce1-495b-bc6e-53637e4c04d4\",\"connector\":{\"name\":\"smooth\"},\"source\":{\"cell\":\"65dcb08a-15c7-4dd3-a46f-f75326bde3e6\",\"port\":\"out\"},\"target\":{\"cell\":\"023d6210-78e1-4bec-979d-53a7ca5abbc6\",\"port\":\"in\"},\"zIndex\":8}]}"
                + "";


        JSONArray cells = JSON.parseObject(jobGraphData).getJSONArray("cells");

        List<JSONObject> nodes = cells.stream().filter(e -> {
            JSONObject cell = (JSONObject) e;
            return cell.getString("shape").equals("job-node");
        }).map(e -> (JSONObject) e).collect(Collectors.toList());

        List<JSONObject> edges = cells.stream().filter(e -> {
            JSONObject cell = (JSONObject) e;
            return cell.getString("shape").equals("edge");
        }).map(e -> (JSONObject) e).collect(Collectors.toList());

        List<DagWorkflowEdge> dagWorkflowEdges = new ArrayList<>(edges.size());
        for (JSONObject edge : edges) {
            DagWorkflowEdge dagWorkflowEdge = new DagWorkflowEdge();
            String sourceId = edge.getJSONObject("source").getString("cell");
            String targetId = edge.getJSONObject("target").getString("cell");

            dagWorkflowEdge.setSourceNodeId(sourceId);
            dagWorkflowEdge.setTargetNodeId(targetId);

            dagWorkflowEdges.add(dagWorkflowEdge);
        }

        List<JSONObject> nodeDataList = nodes.stream().map(e -> e.getJSONObject("data")).collect(Collectors.toList());
        List<DagWorkflowNode> dagWorkflowNodes = new ArrayList<>(nodeDataList.size());
        for (JSONObject jsonObject : nodeDataList) {
            DagWorkflowNode dagWorkflowNode = new DagWorkflowNode();
            dagWorkflowNode.setState(DagWorkflowNodeState.READY);
            dagWorkflowNode.setId(jsonObject.getString("nodeId"));
            dagWorkflowNode.setName(jsonObject.getString("taskName"));
            dagWorkflowNode.setStartTime(LocalDateTime.now());
            dagWorkflowNode.setDagWorkflowNodeTask(new FlinkDagWorkflowTask());

            dagWorkflowNodes.add(dagWorkflowNode);
        }


        ExecutorService executorService = ThreadPoolUtil.getExecutorService();

        DagWorkflow dagWorkflow = new DagWorkflow();
        dagWorkflow.setDagWorkflowEdges(dagWorkflowEdges);
        dagWorkflow.setDagWorkflowNodes(dagWorkflowNodes);
        dagWorkflow.setCreateTime(LocalDateTime.now());
        dagWorkflow.setDescription("");
        dagWorkflow.setVersion("");
        dagWorkflow.setName("11111111");
        dagWorkflow.setId("11111111");

        DagWorkflowExecutor dagWorkflowExecutor = new DagWorkflowExecutor.Builder(dagWorkflow, executorService)
                .build();

        dagWorkflowExecutor.start();

//        try {
//            TimeUnit.SECONDS.sleep(6);
//            dagWorkflowExecutor.pause();
//            TimeUnit.SECONDS.sleep(5);
//            dagWorkflowExecutor.proceed();
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }
    }


}
